package qdu.edu.com.fushanf4.utils;

import org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import java.io.BufferedWriter;
import java.io.DataOutputStream;
import java.io.IOException;
import java.io.OutputStreamWriter;
import java.util.List;

/**
 * 用来辅助测试Job
 */
public class LocalJobRunner {

    /**
     * 用来比较输出文件内容是否符合预期
     * @param outputPath 输出文件路径
     * @param fileName 输出文件名
     * @param expectOutputs 预期的输出文件内容，用字符串数组存储
     * @throws IOException IO异常
     */
    public static void readOutputFileAndCompareToExceptOutput(String outputPath, String fileName, String[] expectOutputs) throws IOException {
        // 读取文件
        List<String> lines = IOUtils.readLines(FileSystem.get(ConfigurationUtil.getConf()).open(new Path(outputPath, fileName)));
        // 比较文件内容是否符合预期
        assert expectOutputs.length == lines.size();
        for (int i=0; i<expectOutputs.length; i++) {
            assert expectOutputs[i].equals(lines.get(i));
        }
    }

    /**
     * 用来删除已存在的文件
     * @param path 第一个参数是输入路径，第二个参数是输出路径
     * @param inputs 输入数据
     * @param inputFileName 输入文件名
     * @throws IOException IO异常
     */
    public static void deleteFileAndWriteInputData(String[] path, String[] inputs, String inputFileName) throws IOException {
        // 删除输入路径和输出路径的文件
        assert path.length == 2;
        ConfigurationUtil.deleteOutputPathAndFiles(path[0]);
        ConfigurationUtil.deleteOutputPathAndFiles(path[1]);

        Configuration conf = ConfigurationUtil.getConf();
        FileSystem fs = FileSystem.get(conf);

        // 创建输入路径的目录
        Path inputPath = new Path(path[0]);
        fs.mkdirs(inputPath);

        // 写入输入数据到本地
        DataOutputStream file = fs.create(new Path(inputPath, inputFileName));
        BufferedWriter bw = new BufferedWriter(new OutputStreamWriter(file));
        for (String input : inputs) {
            bw.write(input);
            bw.newLine();
        }

        // 关闭流
        bw.close();
        fs.close();
    }
}
